package com.tollge.snowflake_registry;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Context;
import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.vertx.core.eventbus.Message;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.mysqlclient.MySQLConnectOptions;
import io.vertx.mysqlclient.MySQLPool;
import io.vertx.sqlclient.PoolOptions;
import io.vertx.sqlclient.Row;
import io.vertx.sqlclient.RowSet;
import io.vertx.sqlclient.Tuple;
import lombok.extern.log4j.Log4j2;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Log4j2
public class FetchSnowflakeVerticle extends AbstractVerticle {

  // Create the pooled client
  MySQLPool pool = null;

  DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

  private Map<String, JsonObject> data = new HashMap<>();

  @Override
  public void init(Vertx vertx, Context context) {
    super.init(vertx, context);

    MySQLConnectOptions connectOptions = new MySQLConnectOptions()
      .setPort(3306)
      .setHost("192.168.3.11")
      .setDatabase("snowflake")
      .setUser("busi_user")
      .setPassword("busi_user");

    // Pool options
    PoolOptions poolOptions = new PoolOptions().setMaxSize(1);

    // Create the pooled client
    pool = MySQLPool.pool(vertx, connectOptions, poolOptions);

    pool.getConnection().compose(connection -> connection
      .preparedQuery("SELECT * FROM registry")
      .execute()
      .eventually(v -> connection.close())
    ).onSuccess(queryR -> {
      if (queryR.size() > 0) {
        for (Row row : queryR) {
          data.put(genKey(row.getString("application_name"), row.getString("ip"), row.getInteger("port")), rowToJsonObject(queryR.columnsNames(), row));
        }
      }
      log.info("init success, size:" + queryR.size());
    }).onFailure(ex -> log.error("init data failed", ex));
  }

  private String genKey(String name, String ip, Integer port) {
    return name + "|" + ip + ":" + port;
  }

  @Override
  public void start() throws Exception {
    vertx.eventBus().<JsonObject>consumer("fetchMachineCode", message -> {
      JsonObject body = message.body();
      log.info(Thread.currentThread().getName() + "|received a message: " + body);

      String searchKey = genKey(body.getString("application"), body.getString("ip"), body.getInteger("port"));
      if (data.containsKey(searchKey)) {
        message.reply(data.get(searchKey));
        return;
      }

      String selectSqlStr = "SELECT * FROM registry WHERE application_name=? and ip=? and port=?";
      Tuple selectTuple = Tuple.of(body.getString("application"), body.getString("ip"), body.getInteger("port"));
      pool.getConnection().compose(connection -> connectAndExec(selectSqlStr, selectTuple, connection)).onSuccess(rows -> {
        // 查到了就返回
        if (rows.size() > 0) {
          returnObj(message, rows);
          return;
        }

        // 没查到就新增
        String insertSqlStr = "INSERT INTO registry (application_name,ip,port,`index`,create_time,is_delete,create_by,update_time,update_by) " + "select ?, ?, ?, IFNULL(max(`index`),0) + 1, now(), 0, 'sys', now(), 'sys' from registry where application_name=?";
        Tuple updateTuple = Tuple.of(body.getString("application"), body.getString("ip"), body.getInteger("port"), body.getString("application"));
        pool.getConnection().compose(connection -> connectAndExec(insertSqlStr, updateTuple, connection)).onSuccess(count -> {
          // 新增完返回结果
          pool.getConnection().compose(connection -> connectAndExec(selectSqlStr, selectTuple, connection)).onSuccess(queryR -> {
            if (queryR.size() > 0) {
              returnObj(message, queryR);
            }
            message.fail(112, "can not find result,req=" + selectTuple);
          }).onFailure(ex -> returnError(ex, message));

        }).onFailure(ex -> returnError(ex, message));

      }).onFailure(ex -> returnError(ex, message));

    });
  }

  private void returnObj(Message<JsonObject> message, RowSet<Row> rows) {
    JsonArray jsonArray = under2Camel(rows);
    JsonObject jsonObject = jsonArray.getJsonObject(0);
    // 记录缓存
    data.put(genKey(jsonObject.getString("applicationName"), jsonObject.getString("ip"), jsonObject.getInteger("port")), jsonObject);

    message.reply(jsonObject);
  }

  private Future<RowSet<Row>> connectAndExec(String sql, Tuple tuple, io.vertx.sqlclient.SqlConnection connection) {
    return connection
      .preparedQuery(sql)
      .execute(tuple)
      .eventually(v -> connection.close());
  }

  private void returnError(Throwable ex, Message<JsonObject> message) {
    log.error(ex);
    message.fail(111, ex.getMessage());
  }

  protected JsonArray under2Camel(RowSet<Row> rs) {
    JsonArray array = new JsonArray();
    List<String> columns = rs.columnsNames();

    for(Row r : rs) {
      JsonObject j = rowToJsonObject(columns, r);
      array.add(j);
    }
    return array;
  }

  private JsonObject rowToJsonObject(List<String> columns, Row r) {
    JsonObject j = new JsonObject();
    for (String cn : columns) {
      if (r.getValue(cn) instanceof LocalDateTime) {
        j.put(toUpper(cn), r.getLocalDateTime(cn).format(formatter));
      } else {
        j.put(toUpper(cn), r.getValue(cn));
      }
    }
    return j;
  }

  private static final byte UPPER_CASE_OFFSET = 'A' - 'a';

  private String toUpper(String cn) {
    int i = cn.indexOf("_");
    if (i < 0) {
      return cn;
    }

    byte[] newStrBytes = new byte[cn.getBytes().length - 1];
    for (int i1 = 0; i1 < cn.getBytes().length; i1++) {
      if (i1 < i) {
        newStrBytes[i1] = cn.getBytes()[i1];
      } else if (i1 == i + 1) {
        newStrBytes[i1 - 1] = (byte)  (cn.getBytes()[i1] + UPPER_CASE_OFFSET);
      } else if (i1 > i) {
        newStrBytes[i1 - 1] = cn.getBytes()[i1];
      }
    }
    return new String(newStrBytes);
  }
}
